package com.peng.rpc.netty.client;


import com.peng.rpc.enumeration.RpcError;
import com.peng.rpc.exception.RpcException;
import com.peng.rpc.netty.util.CommonDecoder;
import com.peng.rpc.netty.util.CommonEncoder;
import com.peng.rpc.serializer.CommonSerializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.*;

/**
 * netty客户端与服务器建立连接的重试器，最终得到建立连接后的channel
 * @author Wonderson
 * @email wpp.ds@foxmail.com
 * @date 2022/3/4 21:18
 */
@Slf4j
public class ChannelProvider {

    private static final Bootstrap bootstrap = initBootstrap();
    private static EventLoopGroup group;
//    private static Channel channel;
//    //最大重试次数
//    private static final int MAX_RETRY_COUNT = 5;

    private static Map<String, Channel> channelMap = new ConcurrentHashMap<>();

    public static Channel getChannel(InetSocketAddress inetSocketAddress, CommonSerializer serializer) {
        //客户端复用已经建立的连接
        String channelName = inetSocketAddress.getAddress().toString() + serializer.getCode();
        if (channelMap.containsKey(channelName)) {
            Channel channel = channelMap.get(channelName);
            if (channel != null && channel.isActive()) {
                return channel;
            } else {
                channelMap.remove(channelName);
            }
        }
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel channel) throws Exception {
                channel.pipeline()
                        .addLast(new CommonEncoder(serializer))
                        .addLast(new CommonDecoder())
//                        .addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS))
                        .addLast(new NettyClientHandler());
            }
        });
        Channel channel = null;
        try {
            channel = connect(bootstrap, inetSocketAddress);
            channelMap.put(channelName, channel);
        } catch (Exception e) {
            log.error("获取连接时发生错误：{}", e.getMessage());
        }
        return channel;
    }


    private static Channel connect(Bootstrap bootstrap, InetSocketAddress inetSocketAddress) throws ExecutionException, InterruptedException {
        CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
        bootstrap.connect(inetSocketAddress)
                .addListener((ChannelFutureListener)future -> {
                    if (future.isSuccess()) {
                        log.info("客户端连接成功：{}：{}", inetSocketAddress.getAddress(), inetSocketAddress.getPort());
                        channelFuture.complete(future.channel());
                    } else {
                        throw new IllegalAccessException("连接获取失败: " + inetSocketAddress.getAddress());
                    }
                });
        return channelFuture.get();
    }

    private static Bootstrap initBootstrap() {
        Bootstrap bootstrap = new Bootstrap();
        group = new NioEventLoopGroup();
        return bootstrap.channel(NioSocketChannel.class)
                .group(group)
                //开启TCP的心跳机制，保证TCP连接的活性
                .option(ChannelOption.SO_KEEPALIVE, true)
                //连接的超时时间，超过这个时长断开连接
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                //开启nagle算法
                .option(ChannelOption.TCP_NODELAY, true);
    }
}
